package com.flute.haflute.master;

import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.List;

import org.jdom.JDOMException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.flute.haflute.agent.AgentContext;
import com.flute.haflute.agent.topo.ClusterTopoDBService;
import com.flute.haflute.agent.topo.Node;
import com.flute.haflute.common.ClusterConstants;
import com.flute.haflute.common.ClusterContext;
import com.flute.haflute.common.ClusterConstants.CLI;
import com.flute.haflute.common.ClusterConstants.ROLE;
import com.flute.haflute.jobbox.base.RunProcessInterface;
import com.flute.haflute.jobbox.common.CloudContext;
import com.flute.haflute.jobbox.common.RunProcessLauncher;
import com.flute.haflute.jobbox.common.tools.OSUtil;
import com.flute.haflute.jobbox.master.JobTracker;
import com.flute.haflute.monitor.GeneralStatusCenter;
import com.flute.haflute.monitor.JobMonitor;
import com.flute.haflute.monitor.JvmReport;
import com.flute.haflute.net.RMINetService;
import com.flute.haflute.tools.ClusterUtils;

/**
 * 任务执行的主控JVM。
 */
public class Master {

	private static Logger logger = LoggerFactory.getLogger("master.Main");
	// 日志存储的配置文件，log4j
	private static String logConfigFile = "conf/log-master.lcf";

	/**
	 * 主线程
	 * 
	 * @param args
	 * @throws Exception
	 */
	public static void main(String[] args) throws Exception {
		try {
			System.out.println("MASTER is starting now, please wait...");

			initContext(args);
			startNetService();
			startMasterBackends();
			starCustomBackends();
			addShutDownHook();
			setDefaultExceptionHander();
			startAgentService(args);
			startJobMonitor();
			new JvmReport(OSUtil.getLocalIP(false) + "-Master");
			GeneralStatusCenter.getInstance().setMyId(
					OSUtil.getLocalIP(false) + "-Master");

			System.out.println("====== MASTER started successfully ======");
			System.out.println(ClusterConstants.PROCESS_STARTED);
			// redirectSystemOut(args);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	private static void startJobMonitor() {
		JobMonitor jobmonitor = new JobMonitor();
		jobmonitor.start();
	}

	/**  */
	private static void setDefaultExceptionHander() {
		// 抓住没有捕获的异常，防止导致JVM崩溃
		Thread.setDefaultUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
			public void uncaughtException(Thread t, Throwable e) {
				logger.error("Fatal exception in thread " + t, e);
				e.printStackTrace(System.err);
				if (e instanceof OutOfMemoryError) {
					ClusterUtils.exitGracefully();
				}
			}
		});
	}

	/**
	 * @param args
	 */
	private static void startAgentService(String[] args) {
		new MasterCommandHandler("Master Command Handler").start();
	}

	private static void addShutDownHook() {
		// 向系统注册Hook，系统关闭时清理环境
		Runtime.getRuntime().addShutdownHook(new ShutdownHook());
	}

	// 检测并启动RMI服务器
	private static void startNetService() {
		RMINetService.getInstance();
	}

	/**
	 * 初始化子节点系统、软件和平台环境
	 * 
	 * @throws IOException
	 * @throws JDOMException
	 */
	private static void initContext(String[] args) throws JDOMException,
			IOException {
		for (String arg : args) {
			if (arg.startsWith(CLI.ROOTDIR_IS)) {
				String rootDir = arg.replaceFirst(CLI.ROOTDIR_IS, "");
				ClusterContext.setRootDir(rootDir);
				CloudContext.setRootDir(rootDir);
				continue;
			}
		}

		ClusterUtils.loadDefaultConfiguration();
		ClusterUtils.configLogger(logConfigFile);
		adjustContext(args);
	}

	private static void adjustContext(String[] args) {
		if (ClusterUtils.isLocalNode(ClusterContext.getDM())) {
			AgentContext.setNodeRole(ROLE.DM);
			String localIp = ClusterContext.getLocalAddress().getHostAddress();
			ClusterTopoDBService.getInstance().setDM(new Node(localIp));
			ClusterContext.setDM(localIp);
			return;
		} else if (ClusterUtils.isLocalNode(ClusterContext.getBDM())) {
			AgentContext.setNodeRole(ROLE.BDM);
			String localIp = ClusterContext.getLocalAddress().getHostAddress();
			ClusterTopoDBService.getInstance().setBDM(new Node(localIp));
			ClusterContext.setBDM(localIp);
			return;
		}
	}

	public static void shutdownProcesses() throws FileNotFoundException,
			InstantiationException, IllegalAccessException {
		for (RunProcessInterface process : RunProcessLauncher
				.getRunProcessList()) {
			process.shutdown();
		}
	}

	/**
	 * 启动用户自定义的后台线程
	 * 
	 * @throws FileNotFoundException
	 * @throws IllegalAccessException
	 * @throws InstantiationException
	 */
	private static void starCustomBackends(String... params) throws Exception,
			IllegalAccessException {
		// ClusterUtils.startCustomedBackends(params);
	}

	/** 启动主节点的后台线程 */
	private static void startMasterBackends(String... params) throws Exception {
		// 所有的后台线程名字记录在案
		String[] coreNames = new String[1];
		coreNames[0] = JobTracker.class.getSimpleName();

		// 真正的后台线程启动
		RunProcessInterface[] coreBackends = new RunProcessInterface[1];
		coreBackends[0] = new JobTracker(); // 后台执行调度线程
		for (RunProcessInterface process : coreBackends) {
			process.callMain(params);
			RunProcessLauncher.addRunProcess2Cache(process);
		}

		// 监控是否所有的后台线程都起来了
		int initialized = 0;
		while (initialized < coreBackends.length) {
			for (int i = 0; i < coreBackends.length; i++) {
				if (coreBackends[i].isInitialized()) {
					initialized++;
					System.out.println("process [" + coreNames[i]
							+ "] started OK");
				}
			}
			ClusterUtils.doWait();
		}
	}

}

/**
 * 系统关闭时，需要清理环境等，该Hook类处理这些事情
 * 
 */
class ShutdownHook extends Thread {
	private static Logger logger = LoggerFactory.getLogger(ShutdownHook.class);

	@Override
	public void run() {
		logger.info("received exit signal, begin to shutdown...");
		List<RunProcessInterface> processList = RunProcessLauncher
				.getRunProcessList();
		for (RunProcessInterface process : processList) {
			process.shutdown();
		}
		try {
			sleep(500L);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		logger.info("shutdown all backend processes OK");
		logger.info("Job Master shutdown OK");
	}

}
